package com.unism.uniMeterage.meterage.stream.version1_0;

import com.unism.uniMeterage.meterage.stream.version1_0.dataWork.Work_0x4C;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author liuc
 * 工作节点管理
 */
public class WorkNodeManager {

    private static  ConcurrentHashMap<String,WorkNode> CacheMapWorker=new ConcurrentHashMap<String,WorkNode>();
    private static WorkNodeManager instance;
    public static WorkNodeManager instance() {
        if (instance == null) {
            // 实例化任务管理
            instance = new WorkNodeManager();
        }
        return instance;
    }

    /**
     * 初始化工作工作节点
     */
    public void initNode(){
        Work_0x4C work_0x4C=new Work_0x4C(this,"4c",1);
        register(work_0x4C);
    }
    /**
     * 在工作节点注册worker
     */
    public void register(WorkNode worker){
      String tempName=worker.getName();
      Thread currThread=worker.currThread();
      ConcurrentLinkedQueue queue=worker.currQueue();
      if(tempName !=null && !tempName.trim().equals("")
              && !CacheMapWorker.contains(tempName) && currThread !=null && queue !=null){
          CacheMapWorker.put(tempName,worker);
      }
    }
    /**
     * 在工作节点删除指定worker
     */
    public void delete(String name){
        WorkNode worker=CacheMapWorker.get(name);
        worker.destory();
        CacheMapWorker.remove(name);
    }
    /**
     * 每个worker,分发数据
     * @param name 目标worker名称
     * @param data  分发的数据
     * @return
     */
    public Boolean distribute(String name,Object data){
        WorkNode currWorker= CacheMapWorker.get(name);

        if(currWorker == null){
            return  false;
        }
       /* Integer currStatus=currWorker.currStatus();
        if(currStatus != 202){
            return  false;
        }*/
        ConcurrentLinkedQueue currQueue= currWorker.currQueue();
        if(currQueue == null){
            return  false;
        }
        currQueue.offer(data);
        return  true;
    }

}
